package ex.tableapi;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/#groupby-window-aggregation
 *
 * 更换jar ：  https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc/3.1.1-1.17
 */
public class Ex07 extends ApiFrame {
    private String outName = "printOutTable";
    private String createPrintOutDDL = "CREATE TABLE printOutTable (" +
            "  `a` STRING," +
            "  `log_ts` TIMESTAMP(3)," +
            "  `amount` BIGINT" +
            ") WITH (" +
            " 'connector' = 'print' " +
            ")";

    public static void main(String[] args) {
        Ex07 demo =new Ex07();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result = orders.window(Tumble.over(lit(1).minute()).on($("rowtime")).as("log_ts"))
                .groupBy($("a"), $("log_ts"))
                .select(
                        $("a"),
                        $("log_ts").start().as("log_ts"),
                        $("b").sum().as("amount"));
        result.printSchema();

        demo.registerConsole(demo.createPrintOutDDL,demo.outName);
        result.executeInsert(demo.outName);
    }
}
